package com.abc.utils;

import com.abc.config.SparkGptConfig;
import com.abc.constant.SparkConstant;
import com.abc.dto.request.ApiRequestRequestDto;
import com.abc.dto.response.SseResponse;
import okhttp3.*;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

@Component
public class SparkRequestUtil {
    private static final Logger logger = LoggerFactory.getLogger(SparkRequestUtil.class);
    private final SparkGptConfig sparkGptConfig;
    private final OkHttpClient client = new OkHttpClient();

    public SparkRequestUtil(SparkGptConfig sparkGptConfig) {
        this.sparkGptConfig = sparkGptConfig;
    }

    /**
     * 发送API请求并使用SseEmitter逐条推送响应
     *
     * @param requestDto 请求参数对象
     * @param emitter    SseEmitter对象，用于推送数据
     */
    public void sendApiStreamRequest(ApiRequestRequestDto requestDto, SseEmitter emitter) {
        MediaType mediaType = MediaType.parse("application/json; charset=utf-8");
        RequestBody requestBody = RequestBody.create(JSON.toJSONString(requestDto), mediaType);

        Request request = new Request.Builder()
                .url(SparkConstant.CHAT_URL)
                .addHeader("Authorization", SparkConstant.AUTHORIZATION + sparkGptConfig.getApiPassword())
                .post(requestBody)
                .build();

        client.newCall(request).enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                logger.error("请求因网络错误失败", e);
                try {
                    emitter.completeWithError(e);
                } catch (Exception ex) {
                    logger.error("SseEmitter 推送失败", ex);
                }
            }

            @Override
            public void onResponse(Call call, Response response) {
                if (response.isSuccessful()) {
                    try (ResponseBody responseBody = response.body()) {
                        if (responseBody != null) {
                            handleSseResponse(responseBody, emitter);
                        }
                    } catch (IOException e) {
                        logger.error("读取SSE响应时出错", e);
                        emitter.completeWithError(e);
                    }
                } else {
                    logger.warn("请求失败，状态码: {}", response.code());
                    emitter.complete();
                }
            }
        });
    }

    /**
     * 逐行读取并推送SSE响应数据
     *
     * @param responseBody 响应体，包含SSE流
     * @param emitter      SseEmitter用于逐条推送数据
     */
    private void handleSseResponse(ResponseBody responseBody, SseEmitter emitter) throws IOException {
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(responseBody.byteStream()))) {
            String line;
            while ((line = reader.readLine()) != null) {
                logger.info("收到SSE数据: {}", line);
                processSseData(line, emitter);
            }
            emitter.complete();
        } catch (IOException e) {
            logger.error("处理SSE数据时出错", e);
            emitter.completeWithError(e);
            throw e;
        }
    }

    /**
     * 处理接收到的每行SSE数据，将解析后的数据推送到SseEmitter
     *
     * @param data    从SSE流中接收到的数据行
     * @param emitter 用于推送数据的SseEmitter
     */
    private void processSseData(String data, SseEmitter emitter) {
        if (data.startsWith("data: ")) {
            String jsonData = data.substring(6).trim();

            if ("[DONE]".equals(jsonData)) {
                logger.info("SSE 数据流已结束");
                return;
            }

            try {
                SseResponse responsePart = JSON.parseObject(jsonData, SseResponse.class);
                emitter.send(responsePart);  // 逐条推送数据到客户端
            } catch (Exception e) {
                logger.error("解析或推送 JSON 数据时出错", e);
                emitter.completeWithError(e);
            }
        }
    }
}
